package com.avo.ramcache.persist.disruptor;

import com.avo.ramcache.config.IEntity;
import com.avo.ramcache.orm.Accessor;
import com.avo.ramcache.persist.Element;
import com.avo.ramcache.persist.Listener;
import com.avo.ramcache.persist.Persister;
import com.avo.ramcache.persist.disruptor.event.ElementEvent;
import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;

public class DisruptorPersister implements Persister {

    private static final Logger LOGGER = LoggerFactory.getLogger(DisruptorPersister.class);
    private boolean initialize;
    private volatile boolean stop;
    private DisruptorConsumer consumer;
    private RingBuffer<ElementEvent> ringBuffer;
    private final ConcurrentHashMap<Class<? extends IEntity<?>>, Listener> listeners = new ConcurrentHashMap<>();

    @Override
    public void initialize(String name, Accessor accessor, String config) {
        if (initialize) {
            return;
        }
        ExecutorService executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS, new SynchronousQueue<>(), runnable -> new Thread(runnable, name + "-thread"));
        int bufferSize = Math.max(1, Integer.parseInt(config));
        ElementEventFactory factory = new ElementEventFactory();
        Disruptor<ElementEvent> disruptor = new Disruptor<>(factory, bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());
        consumer = new DisruptorConsumer(name, accessor, disruptor, executor, this);
        disruptor.handleEventsWith(consumer);
        disruptor.handleExceptionsFor(consumer).with(new ExceptionHandler<ElementEvent>() {
            @Override
            public void handleEventException(Throwable ex, long sequence, ElementEvent event) {
                LOGGER.error(name + "disruptor 异常", ex);
            }

            @Override
            public void handleOnStartException(Throwable ex) {
                LOGGER.error(name + "disruptor 异常", ex);
            }

            @Override
            public void handleOnShutdownException(Throwable ex) {
                LOGGER.error(name + "disruptor 异常", ex);
            }
        });
        disruptor.start();
        ringBuffer = disruptor.getRingBuffer();
        initialize = true;
    }

    @Override
    public void put(Element element) {
        if (stop) {
            return;
        }
        long next = ringBuffer.next();
        ElementEvent elementEvent = ringBuffer.get(next);
        elementEvent.setElement(element);
        ringBuffer.publish(next);
    }

    @Override
    public void addListener(Class<? extends IEntity<?>> clz, Listener listener) {
        if (listener == null) {
            throw new RuntimeException("监听器为空");
        }
        this.listeners.put(clz, listener);
    }

    @Override
    public Listener getListener(Class<? extends IEntity<?>> clz) {
        return this.listeners.get(clz);
    }

    @Override
    public void shutdown() {
        this.stop = true;
        consumer.stop();
    }

    @Override
    public Map<String, String> getInfo() {
        HashMap<String, String> result = new HashMap<>();
        result.put("bufferSize", String.valueOf(ringBuffer.getBufferSize()));
        result.put("cursor", String.valueOf(ringBuffer.getCursor()));
        result.put("minimumGatingSequence", String.valueOf(ringBuffer.getMinimumGatingSequence()));
        return result;
    }
}
